/** sort merge join algorithm **/

package qp.operators;

import qp.utils.*;
import java.io.*;
import java.util.*;
import java.lang.*;
import qp.optimizer.*;
import java_cup.internal_error;

//public class SortMergeJoin extends Join{
public class ExternalMergeSortJoin extends Join{
     static int stage=0;
    static int IOCost=0;
    static int filenum=0;   
    int batchsize;  
    int leftindex;
    int rightindex;
    String lfname;	
    String rfname;  
  
    Batch outbatch=null;   
    Batch anotheroutbath=null;
    Batch leftBatch=null;
    Batch rightBatch=null;
	
    Tuple leftTuple;
    Tuple rightTuple;
    List<Tuple> tempLeft;
    List<Tuple> tempRight;
    ObjectInputStream leftois, rightois;
    
    public ExternalMergeSortJoin(Join jn){
		super(jn.getLeft(),jn.getRight(),jn.getCondition(),jn.getOpType());
		schema = jn.getSchema();
		jointype = jn.getJoinType();
		numBuff = BufferManager.getBuffersPerJoin();	
    }

	private Tuple NextLeftTuple() {
		if (leftBatch.isEmpty()) {	    
			try{			
				leftBatch = (Batch)leftois.readObject();
				IOCost++;
			}
			catch (java.io.EOFException e)
			{
				leftBatch=null;
				return null;
			}catch (Exception g)
			{
			}
	
		}
		return NextTuple(leftBatch);
    }

    private Tuple NextRightTuple() {
		if (rightBatch.isEmpty()) {
			try{			
				rightBatch = (Batch)rightois.readObject();
				IOCost++;
			}catch (java.io.EOFException f)
			{
				rightBatch = null;
				return null;
			}catch (Exception g)
			{
			}
		}	
		return NextTuple(rightBatch);
    }
			
    private Tuple NextTuple(Batch batch) {
		if (batch != null) {
			Tuple tuple = batch.getTuples().get(0);
			batch.getTuples().remove(0);
			return tuple;
		} else {
			return null;
		}
    }

    public String sortInput(Operator op,Attribute attr) {
    	int i=0;
    	if(!op.open()) {   		
    		return null;
    	}
    	else {
    		filenum++;
    		String unSortedFile="unSortedFile-"+String.valueOf(filenum);
    		try {
    			FileOutputStream fos=new FileOutputStream(unSortedFile);
    			ObjectOutputStream oos=new ObjectOutputStream(fos);
    			
    			int tupleSize=op.getSchema().getTupleSize();
    			int batchSize=Batch.getPageSize()/tupleSize;
    			Batch tempOutputBatch=new Batch(batchSize);
    			    	
    			while((tempOutputBatch=op.next())!=null) {
    				int tempSize=tempOutputBatch.size();
    				i=i+tempSize;   	
    				oos.writeObject(tempOutputBatch);
    				tempOutputBatch=new Batch(batchSize);
    			}
    			fos.close();
    			oos.close();
    		}
    		catch (Exception e) {
    			e.printStackTrace();
    		}
    		
    		String sortedFile = null;
    		Vector<Attribute> attrVector=new Vector<Attribute>();//convert attr to Vector
    		attrVector.add(attr);
    		
			ExternalMergeSort es=new ExternalMergeSort(unSortedFile,attrVector,op.getSchema());
    		try {
    			sortedFile=es.MergeSort();			
    		}
    		catch (ArrayIndexOutOfBoundsException e) {
				e.printStackTrace();				
			}
    		catch (Exception e) {    			
    			e.printStackTrace();
    		}	
    		return sortedFile;
    	}
    }


    public boolean open(){

		int tuplesize=schema.getTupleSize();
		batchsize=Batch.getPageSize()/tuplesize;

		Attribute leftattr = con.getLhs();
		Attribute rightattr =(Attribute) con.getRhs();
		
		leftindex = left.getSchema().indexOf(leftattr);
		rightindex = right.getSchema().indexOf(rightattr);
		
		
		
		if(!left.open() || !right.open()) {
			return false;
		}
		else {		
			lfname=this.sortInput(left, leftattr);			
			rfname=this.sortInput(right, rightattr);
		}
		if(lfname==null || rfname==null) {
			return false;
		}
		else {
			try {
				FileInputStream leftfis=new FileInputStream(lfname);
				FileInputStream rightfis=new FileInputStream(rfname);
				leftois=new ObjectInputStream(leftfis);
				rightois=new ObjectInputStream(rightfis);
			}
			catch (Exception e) {				
				e.printStackTrace();
			}
			leftBatch=new Batch(batchsize);
			rightBatch=new Batch(batchsize);

			try{			
				leftBatch = (Batch)leftois.readObject();
			}
			catch (java.io.EOFException e)
			{
				leftBatch=null;
			}catch (Exception g)
			{
			}

			try{			
				rightBatch = (Batch)rightois.readObject();
			}
			catch (java.io.EOFException e)
			{
				rightBatch=null;
			}catch (Exception g)
			{
			}
			
			tempLeft = new ArrayList<Tuple>();
			tempRight = new ArrayList<Tuple>();

			leftTuple=NextTuple(leftBatch);
			rightTuple=NextTuple(rightBatch);
			
			anotheroutbath=new Batch(batchsize);

			return true;
		}
    }
  
    public Batch next() {
    	
		if (leftBatch==null || rightBatch==null) {
			return null;
		}

		outbatch = new Batch(batchsize);

		while (!outbatch.isFull() && !anotheroutbath.isEmpty()) {
			outbatch.add(anotheroutbath.elementAt(0));
			anotheroutbath.remove(0);
		}


		while (!outbatch.isFull() && leftBatch!=null && rightBatch!=null) {

			int compare = Tuple.compareTuples(leftTuple, rightTuple, leftindex, rightindex);

			if (compare == 0) {
				tempLeft.add(leftTuple);
				tempRight.add(rightTuple);

				rightTuple = NextRightTuple();
				while (rightTuple != null && Tuple.compareTuples(tempLeft.get(0), rightTuple, leftindex, rightindex) == 0) {
					tempRight.add(rightTuple);
					rightTuple =NextRightTuple();
				}
				
				leftTuple = NextLeftTuple();
				while (leftTuple != null && Tuple.compareTuples(leftTuple, tempRight.get(0), leftindex, rightindex) == 0) {
					tempLeft.add(leftTuple);
					leftTuple = NextLeftTuple();
				}

				for (Tuple leftTuple : tempLeft) {
					for (Tuple rightTuple : tempRight) {
						if (!outbatch.isFull()) {
							outbatch.add(leftTuple.joinWith(rightTuple));
						} else {
							anotheroutbath.add(leftTuple.joinWith(rightTuple));
						}
					}	
				}
	
				tempRight = new ArrayList<Tuple>();
				tempLeft = new ArrayList<Tuple>();

			} else if (compare < 0) {
				leftTuple = NextLeftTuple();
			} else if (compare > 0) {
				rightTuple =NextRightTuple();
			}

		}
		IOCost++;
		return outbatch;
    }
    
    
    /** Close the operator */
    public boolean close(){
		File l = new File(lfname);
		File f = new File(rfname);
		l.delete();
		f.delete();
		System.out.println("Stage@ "+stage+" IOCost in SJ is "+IOCost);
		stage++;
		return true;

    }

}
